package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryPrefixComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;

public class Demo5Index {
    Connection connection;

    /**
     * 创建链接
     */
    @Before
    public void init() {
        //1、创建hbase链接
        //创建hadoop配置对象
        Configuration configuration = HBaseConfiguration.create();
        //指定链接参数
        //指定zk地址，在zk中保存了hbase元数据
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
        //建立链接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 二级索引，
     * 1、将列值放到rowkey中
     * <p>
     * create 'student_index','info'
     */
    @Test
    public void putStudent() throws Exception {

        Table studentIndex = connection.getTable(TableName.valueOf("student_index"));

        ArrayList<Put> puts = new ArrayList<>();

        BufferedReader bufferedReader = new BufferedReader(new FileReader("../data/students.txt"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            String age = split[2];
            String gender = split[3];
            String clazz = split[4];

            //构建rowkey
            String rowkey = gender + "_" + id;

            //构建put
            Put put = new Put(Bytes.toBytes(rowkey));

            //增加列
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(name));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("age"), Bytes.toBytes(age));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("gender"), Bytes.toBytes(gender));
            put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("clazz"), Bytes.toBytes(clazz));

            puts.add(put);
        }


        //插入数据
        studentIndex.put(puts);
    }

    /**
     * 查询二级索引表
     * 1、使用rowkey前缀过滤器查询
     *
     * @throws Exception
     */
    @Test
    public void queryByGender() throws Exception {
        Table studentIndex = connection.getTable(TableName.valueOf("student_index"));

        //构建前缀比较器
        BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator(Bytes.toBytes("男"));

        //创建行键过滤器
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

        //创建scan
        Scan scan = new Scan();

        //增加过滤器
        scan.setFilter(rowFilter);

        //查询数据
        ResultScanner resultScanner = studentIndex.getScanner(scan);

        //解析数据
        for (Result result : resultScanner) {
            //去除rowkey
            String rowkey = Bytes.toString(result.getRow());

            //取出id
            String[] split = rowkey.split("_");
            String id = split[1];

            //解析数据
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            String age = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("age")));
            String gender = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("gender")));
            String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));
            System.out.println(rowkey + "\t" + id + "\t" + name + "\t" + age + "\t" + gender + "\t" + clazz);
        }

    }


    @After
    public void close() {
        try {
            connection.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}
